package com.mmc.delay.container;


import com.mmc.delay.bean.Job;
import com.mmc.delay.bean.TopicAndId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;

import java.util.Set;

import static com.mmc.delay.constants.RedisKeyConstants.REDIS_DELAY_BUCKET;


/**
 * 是一个有序队列
 */
public class DelayBucket {


    private RedisTemplate redisTemplate;

    public DelayBucket(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    private static final Logger logger = LoggerFactory.getLogger(DelayBucket.class);


    private BoundZSetOperations getBucket(String partition) {
        return redisTemplate.boundZSetOps(REDIS_DELAY_BUCKET.concat(partition));
    }



    public void addDelayJob(String partition,Job job) {
        logger.info("任务移入Delay Set:{}", job.toString());
        TopicAndId topicAndId = new TopicAndId(job.getTopic(), job.getJobId());
        getBucket(partition).add(topicAndId.getTopicAndId(),job.getShoudRunTime());
    }

    public Set<ZSetOperations.TypedTuple> range(String partition, Long count){
        return getBucket(partition).rangeWithScores(0L,count-1);
    }

    public Set<ZSetOperations.TypedTuple> range(String partition,Long start,Long end){
        return getBucket(partition).rangeWithScores(start,end);
    }

    public Long remove(String partition,String topicAndIdStr){
       return getBucket(partition).remove(topicAndIdStr);
    }

    public Long totalNum(String topic){
        return null;
    }









}
